package com.haibara.codesandbox.kafka.listener;

import cn.hutool.json.JSONUtil;
import com.haibara.codesandbox.constant.KafkaConstant;
import com.haibara.codesandbox.controller.Dashboard;
import com.haibara.codesandbox.core.DefaultCodeSandbox;
import com.haibara.codesandbox.kafka.processor.Processor;
import com.haibara.codesandbox.model.ExecuteCodeRequest;
import com.haibara.codesandbox.model.ExecuteCodeRespond;
import com.haibara.codesandbox.model.enums.JudgeResultEnum;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

/**
 * 运行代码Listener
 *
 * @Author: mengyu
 * @Date: 2025/2/11
 */
@Component
@Slf4j
public class RunCodeListener {
    @Resource
    private RetryTemplate kafkaRetryTemplate;
    @Resource
    private DefaultCodeSandbox defaultCodeSandbox;
    @Resource
    private Processor processor;
    @Resource
    private Dashboard dashboard;

    @KafkaListener(topics = KafkaConstant.RUN_CODE_TOPIC, groupId = KafkaConstant.RUN_CODE_GROUP)
    public void consumerJudgeMessage(String content, Acknowledgment ack) {
        log.info("接收到运行请求：{}", content);
        ExecuteCodeRequest ecr = JSONUtil.toBean(content, ExecuteCodeRequest.class);
        kafkaRetryTemplate.execute(retryContext -> {
            try {
                defaultCodeSandbox.executeCode(ecr);
                ack.acknowledge();
                dashboard.count();
            } catch (InterruptedException | ExecutionException e) {
                int retryCount = retryContext.getRetryCount();
                log.error("代码运行请求处理报错，重试次数:{}，题目提交id:{}，msg{}:", retryCount, content, e.getMessage());
                throw new RuntimeException(e);
            }
            return null;
        }, retryContext -> {
            log.error("执行代码发生错误: ", retryContext.getLastThrowable());
            // 发送包含系统错误的执行结果
            ExecuteCodeRespond result = ExecuteCodeRespond.builder()
                    .outputList(null)
                    .message(JudgeResultEnum.SYSTEM_ERROR.getText())
                    .resultStatus(ExecuteCodeRespond.SYSTEM_ERROR)
                    .memoryUsed(0L)
                    .timeUsed(0L)
                    .questionSubmitId(ecr.getQuestionSubmitId())
                    .build();
            processor.sendMsg(KafkaConstant.JUDGE_TOPIC, JSONUtil.toJsonStr(result));
            ack.acknowledge();
            return null;
        });
    }
}